package com.edu.control.config;

import com.edu.control.deploy.retrytask.RetryTask;
import com.edu.control.deploy.retrytask.RetryTaskManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class RetryQueueStarter {

    private static final Logger LOGGER = LoggerFactory.getLogger(RetryQueueStarter.class);

    @Value("${retry.thread.corePoolSize:2}")
    private Integer retryThreadCorePoolSize;
    @Value("${retry.thread.maxPoolSize:2}")
    private Integer retryThreadMaxPoolSize;
    @Value("${retry.thread.queueCapacity:50}")
    private Integer retryQueueCapacity;
    @Value("${retry.thread.name.prefix:retry-}")
    private String retryThreadNamePrefix;
    @Value("${retry.thread.keep.alive.seconds:60}")
    private Integer retryKeepAliveSeconds;

    private ThreadPoolExecutor retryQueuePool;

    private static RetryQueueStarter instance;

    public static RetryQueueStarter getInstance() {
        return instance;
    }

    @Resource
    public void setApplicationContext(ApplicationContext applicationContext) {
        instance = applicationContext.getBean(RetryQueueStarter.class);
    }

    /**
     * 初始化队列监听
     *
     * @throws Exception
     */
    @PostConstruct
    public void init() {
        retryQueuePool = new ThreadPoolExecutor(retryThreadCorePoolSize, retryThreadMaxPoolSize, retryKeepAliveSeconds, TimeUnit.SECONDS
                , new ArrayBlockingQueue<Runnable>(retryQueueCapacity), new ThreadFactory() {
            final AtomicInteger threadNumber = new AtomicInteger(1);

            public Thread newThread(Runnable r) {
                Thread t = new Thread(r,
                        retryThreadNamePrefix + threadNumber.getAndIncrement());
                if (t.getPriority() != Thread.NORM_PRIORITY) {
                    t.setPriority(Thread.NORM_PRIORITY);
                }
                return t;
            }
        }, new ThreadPoolExecutor.CallerRunsPolicy());
        startThread();
    }

    /**
     * 销毁线程池
     *
     * @throws Exception
     */
    public void destroy() {
        if (retryQueuePool != null) {
            retryQueuePool.shutdown();
            storeTask();
        }
    }

    public void storeTask() {
        Object[] objects = RetryTaskManager.retryTaskQueue.toArray();
        for (int i = 0; i < objects.length; i++) {
            RetryTask retryTask = (RetryTask) objects[i];
            if (retryTask != null) {
                retryTask.interrupt();
            }
        }
    }

    private void startThread() {
        LOGGER.info("init retry task delay queue ...");
        final DelayQueue<RetryTask> retryTaskQueue = RetryTaskManager.retryTaskQueue;

        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        if (!isShutdown()) {
                            RetryTask task = retryTaskQueue.take();
                            if (task != null) {
                                if (!isShutdown()) {
                                    retryQueuePool.submit(task);
                                } else {
                                    LOGGER.info("retry thread pool is shutdown, task: {}", task.toString());
                                    task.interrupt();
                                }
                            }
                        } else {
                            break;
                        }
                    } catch (Exception e) {
                        LOGGER.error("task exec error: {}", e.getMessage());
                    }
                }
            }
        }).start();
    }

    public boolean isShutdown() {
        return this.retryQueuePool.isShutdown();
    }
}
